package com.zhao.sparkstreaming.sgg.code.mock

import java.util.Properties

import com.zhao.sparkstreaming.sgg.code.{CityInfo, RanOpt, RandomOptions}
import com.zhao.sparkstreaming.sgg.utils.PropertiesUtil
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import scala.collection.mutable.ArrayBuffer
import scala.util.Random

/**
 * Description: 造数脚本<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/4 14:25
 *
 * @author 柒柒
 * @version : 1.0
 */

object MockRealTime {

  /**
   * 模拟的数据
   *
   * 格式: timestamp area city userid adid
   * 某个时间点 某个地区 某个城市 某个用户 某个广告
   */
  def generateMockData(): Array[String] = {
    val array: ArrayBuffer[String] = ArrayBuffer[String]()
    val CityRandomOpt: RandomOptions[CityInfo] = RandomOptions(RanOpt(CityInfo(1, "北京", "华北"), 30),
      RanOpt(CityInfo(2, "上海", "华东"), 30),
      RanOpt(CityInfo(3, "广州", "华南"), 30),
      RanOpt(CityInfo(4, "深圳", "华南"), 30),
      RanOpt(CityInfo(5, "天津", "华北"), 30)
    )

    val random: Random = new Random()
    //模拟实时数据
    // timestamp area city userid adid
    for (i <- 0 to 50) {
      val timestamp: Long = System.currentTimeMillis()
      val cityInfo: CityInfo = CityRandomOpt.getRandomOpt
      val city: String = cityInfo.city_name
      val area: String = cityInfo.area
      val adid: Int = 1 + random.nextInt(6)
      val userid: Int = 1 + random.nextInt(6)

      //拼接实时数据
      array += timestamp + " " + area + " " + city + " " + userid + " " + adid
    }
    array.toArray
  }

  def createKafkaProducer(broker: String) = {

    //创建配置对象
    val prop = new Properties()
    //添加配置
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")

    //根据配置创建Kafka
    new KafkaProducer[String, String](prop)
  }

  def main(args: Array[String]): Unit = {

    //获取配置文件config.properties 中的kafka配置参数
    val config: Properties = PropertiesUtil.load("config.properties")
    val broker: String = config.getProperty("kafka.broker.list")
    val topic = "test"

    //创建kafka消费者
    val kafkaProducer: KafkaProducer[String, String] = createKafkaProducer(broker)

    while (true) {
      //随机产生实时数据并通过kafka生产者发送到Kafka集群中
      for (line <- generateMockData()) {
        kafkaProducer.send(new ProducerRecord[String, String](topic, line))
        println(line)
      }
      Thread.sleep(2000)
      println("---------------------")
    }
  }
}
